package cc.ttiot.service.dataTransmission.mqtt;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import static cc.ttiot.enums.SystemConstant.*;

/**
 * @Description: 用于保存MQTT客户端信息
 * 在主程序解决异常,这里仅仅抛出
 * @Author :
 * @Version :
 * @Date :
 */

//参考以下文章
//https://mcxiaoke.gitbooks.io/mqtt-cn/content/    mqtt协议中文版
//https://blog.csdn.net/u012546338/article/details/75048847
//https://blog.csdn.net/yao_guai/article/details/90601260
public class Client {
    private static volatile Client client;

    /**
     * @field user          连接MQTT服务器的用户名
     * @field password      连接MQTT服务器的密码
     * @field clientID      MQTT客户端的唯一标识符（不能与该服务器下其他客户端的clientID重复）
     * @field mqttClient    MQTT客户端
     */

    private MqttClient mqttClient;
    private MqttConnectOptions connOpts;

    {
        /*
         * 创建MqttClient客户端
         */
        //内存存储
        MemoryPersistence persistence = new MemoryPersistence();
        //clientId
        String clientId = "pd" + DOOR_ID;

        try {
            mqttClient = new MqttClient(BROKER, clientId, persistence);
        } catch (MqttException e) {
            e.printStackTrace();
        }

        // 设置回调函数
        mqttClient.setCallback(new MqttCallback() {

            //连接丢失时执行重连
            public void connectionLost(Throwable cause) {

                System.out.println("connectionLost");

                if (!client.isConnected()) {
                    try {
                        client.connect();
                    } catch (MqttException e) {
                        e.printStackTrace();
                    }
                    System.out.println("reconnected");
                }

            }

            //收到消息时执行
            public void messageArrived(String topic, MqttMessage message) throws Exception {
//                String mes = new String(message.getPayload());
//                System.out.println("收到来自 " + topic + " 主题的消息！");
//                System.out.println("采集到的数据为 " + mes);
            }

            //发送完成时执行
            public void deliveryComplete(IMqttDeliveryToken token) {
//                System.out.println("deliveryComplete---------" + token.isComplete());
            }

        });

        /*
         * 设置连接参数connOpts
         */
        //创建链接参数
        connOpts = new MqttConnectOptions();
        //设置连接的用户名
        connOpts.setUserName(MQTT_USER_NAME);
        //设置连接的密码
        connOpts.setPassword(MQTT_PASSWORD.toCharArray());
        //设置连接超时时间,单位为秒
        connOpts.setConnectionTimeout(60);
        //设置会话心跳时间,单位为秒,服务器会每隔(1.5 * parameter)秒的时间向客户端发送消息判断客户端是否在线,但不会重连
        //设置一个以秒为单位的时间间隔，Client和Broker之间在这个时间间隔之内需要至少一次消息交互，否则Client和Broker会认为它们之间的连接已经断开。
        //目前设置是半小时没有数据传输自动断连
        connOpts.setKeepAliveInterval(20 * 60);
        //在重新启动和重新连接时记住状态
        connOpts.setCleanSession(true);
        //设置自动重连
//        connOpts.setAutomaticReconnect(true);
    }

    //构造函数
    private Client() {
        super();
    }

    //获取Client对象
    public static Client getInstance() {
        if (client == null) {
            synchronized (Client.class) {
                if (client == null) {
                    client = new Client();
                }
            }
        }
        return client;
    }

    //创建与MQTT服务器的连接
    public synchronized void connect() throws MqttException {
        //未连接
        if (!mqttClient.isConnected()) {
            // 建立连接
            mqttClient.connect(connOpts);
        }
    }

    //关闭与MQTT服务器的连接
    public synchronized void disconnect() throws MqttException {
        //连接着
        if (mqttClient.isConnected()) {
            // 断开连接
            mqttClient.disconnect();
        }
    }

    //是否已连接
    public boolean isConnected() {
        return mqttClient.isConnected();
    }

    //关闭客户端(不要调用该方法)
    public void closeClient() throws MqttException {
        // 关闭客户端
        mqttClient.close();
    }

    //发布消息
    public void publishMessage(String topic, String message) throws MqttException {
        // 创建消息
        MqttMessage mqttMessage = new MqttMessage(message.getBytes());
        // 发布消息
        mqttClient.publish(topic, mqttMessage);
    }

    public void publishMessage(String topic, String message, int qos) throws MqttException {
        // 创建消息
        MqttMessage mqttMessage = new MqttMessage(message.getBytes());
        // 设置消息的服务质量
        mqttMessage.setQos(qos);
        // 发布消息
        mqttClient.publish(topic, mqttMessage);
    }

    //订阅消息(打印)
    public void subscribeMessage(String topic) throws MqttException {
        // 设置回调函数
        mqttClient.setCallback(new MqttCallback() {

            public void connectionLost(Throwable cause) {//当mosquito服务器挂了之后,就爆出这个信息
                System.out.println("connectionLost");
            }

            //收到消息时调用该方法
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                String mes = new String(message.getPayload());
                System.out.println("收到来自 " + topic + " 主题的消息！");
                System.out.println("采集到的数据为 " + mes);
            }

            public void deliveryComplete(IMqttDeliveryToken token) {
                System.out.println("deliveryComplete---------" + token.isComplete());
            }

        });
        mqttClient.subscribe(topic);
    }

    public void subscribeMessage(String topic, int qos) throws MqttException {
        // 设置回调函数
        mqttClient.setCallback(new MqttCallback() {

            public void connectionLost(Throwable cause) {
                System.out.println("connectionLost");
            }

            //收到消息时调用该方法
            public void messageArrived(String topic, MqttMessage message) throws MqttException {
                String mes = new String(message.getPayload());
                System.out.println("收到来自 " + topic + " 主题的消息！");
                System.out.println("采集到的数据为 " + mes);

            }

            public void deliveryComplete(IMqttDeliveryToken token) {
                System.out.println("deliveryComplete---------" + token.isComplete());
            }

        });
        mqttClient.subscribe(topic, qos);
    }


}
